package com.central.scheduler.service.registry;

import com.central.scheduler.common.Constants;
import com.central.scheduler.common.IStoppable;
import com.central.scheduler.common.enums.NodeType;
import com.central.scheduler.spi.register.*;
import com.central.scheduler.common.utils.PropertyUtils;
import com.central.scheduler.spi.plugin.PluginLoader;
import com.central.scheduler.spi.plugin.PluginManagerConfig;
import com.google.common.collect.ImmutableList;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;

import static com.central.scheduler.common.Constants.*;

/**
 * @author Tindy
 * @date 2021/10/29
 * @describe All business parties use this class to access the registry
 */
public class RegistryCenter {

    private static final Logger logger = LoggerFactory.getLogger(RegistryCenter.class);

    private final AtomicBoolean isStarted = new AtomicBoolean(false);

    private Registry registry;

    private IStoppable stoppable;

    /**
     * nodes namespace
     */
    protected static String NODES;

    private RegistryPluginManager registryPluginManager;

    protected static final String EMPTY = "";

    private static final String REGISTRY_PREFIX = "registry";

    private static final String REGISTRY_PLUGIN_BINDING = "registry.plugin.binding";

    private static final String REGISTRY_PLUGIN_DIR = "registry.plugin.dir";

    private static final String MAVEN_LOCAL_REPOSITORY = "maven.local.repository";

    private static final String REGISTRY_PLUGIN_NAME = "plugin.name";

    /**
     * default registry plugin dir
     */
    private static final String REGISTRY_PLUGIN_PATH = "lib/plugin/registry";

    private static final String REGISTRY_CONFIG_FILE_PATH = "/registry.properties";

    /**
     * init node persist
     */
    public void init() {
        if (isStarted.compareAndSet(false, true)) {
            PropertyUtils.loadPropertyFile(REGISTRY_CONFIG_FILE_PATH);
            Map<String, String> registryConfig = PropertyUtils.getPropertiesByPrefix(REGISTRY_PREFIX);

            if (null == registryConfig || registryConfig.isEmpty()) {
                throw new RegistryException("registry config param is null");
            }
            if (null == registryPluginManager) {
                installRegistryPlugin(registryConfig.get(REGISTRY_PLUGIN_NAME));
                registry = registryPluginManager.getRegistry();
            }

            registry.init(registryConfig);
            initNodes();

        }
    }

    /**
     * init nodes
     */
    private void initNodes() {
        persist(REGISTRY_SCHEDULER_MASTERS, EMPTY);
        persist(REGISTRY_SCHEDULER_WORKERS, EMPTY);
        persist(REGISTRY_SCHEDULER_DEAD_SERVERS, EMPTY);
    }

    /**
     * install registry plugin
     */
    private void installRegistryPlugin(String registryPluginName) {
        PluginManagerConfig registryPluginManagerConfig = new PluginManagerConfig();
        registryPluginManagerConfig.setPlugins(PropertyUtils.getString(REGISTRY_PLUGIN_BINDING));
        if (StringUtils.isNotBlank(PropertyUtils.getString(REGISTRY_PLUGIN_DIR))) {
            registryPluginManagerConfig.setInstalledPluginsDir(PropertyUtils.getString(REGISTRY_PLUGIN_DIR, REGISTRY_PLUGIN_PATH).trim());
        }

        if (StringUtils.isNotBlank(PropertyUtils.getString(MAVEN_LOCAL_REPOSITORY))) {
            registryPluginManagerConfig.setMavenLocalRepository(PropertyUtils.getString(MAVEN_LOCAL_REPOSITORY).trim());
        }

        registryPluginManager = new RegistryPluginManager(registryPluginName);

        PluginLoader registryPluginLoader = new PluginLoader(registryPluginManagerConfig, ImmutableList.of(registryPluginManager));
        try {
            registryPluginLoader.loadPlugins();
        } catch (Exception e) {
            throw new RuntimeException("Load registry Plugin Failed !", e);
        }
    }

    /**
     * close
     */
    public void close() {
        if (isStarted.compareAndSet(true, false) && registry != null) {
            registry.close();
        }
    }

    public void persist(String key, String value) {
        registry.persist(key, value);
    }

    public void persistEphemeral(String key, String value) {
        registry.persistEphemeral(key, value);
    }

    public void remove(String key) {
        registry.remove(key);
    }

    public void update(String key, String value) {
        registry.update(key, value);
    }

    public String get(String key) {
        return registry.get(key);
    }

    public void subscribe(String path, SubscribeListener subscribeListener) {
        registry.subscribe(path, subscribeListener);
    }

    public void addConnectionStateListener(RegistryConnectListener registryConnectListener) {
        registry.addConnectionStateListener(registryConnectListener);
    }

    public boolean isExisted(String key) {
        return registry.isExisted(key);
    }

    public boolean getLock(String key) {
        return registry.acquireLock(key);
    }

    public boolean releaseLock(String key) {
        return registry.releaseLock(key);
    }

    /**
     * @return get dead server node parent path
     */
    public String getDeadZNodeParentPath() {
        return REGISTRY_SCHEDULER_DEAD_SERVERS;
    }

    public void setStoppable(IStoppable stoppable) {
        this.stoppable = stoppable;
    }

    public IStoppable getStoppable() {
        return stoppable;
    }

    /**
     * whether master path
     *
     * @param path path
     * @return result
     */
    public boolean isMasterPath(String path) {
        return path != null && path.contains(REGISTRY_SCHEDULER_MASTERS);
    }

    /**
     * get worker group path
     *
     * @param workerGroup workerGroup
     * @return worker group path
     */
    public String getWorkerGroupPath(String workerGroup) {
        return REGISTRY_SCHEDULER_WORKERS + "/" + workerGroup;
    }

    /**
     * whether worker path
     *
     * @param path path
     * @return result
     */
    public boolean isWorkerPath(String path) {
        return path != null && path.contains(REGISTRY_SCHEDULER_WORKERS);
    }

    /**
     * get children nodes
     *
     * @param key key
     * @return children nodes
     */
    public List<String> getChildrenKeys(final String key) {
        return registry.getChildren(key);
    }

    /**
     * opType(add): if find dead server , then add to zk deadServerPath
     * opType(delete): delete path from zk
     *
     * @param node node path
     * @param nodeType master or worker
     * @param opType delete or add
     * @throws Exception errors
     */
    public void handleDeadServer(String node, NodeType nodeType, String opType) throws Exception {
        String host = getHostByEventDataPath(node);
        String type = (nodeType == NodeType.MASTER) ? MASTER_TYPE : WORKER_TYPE;

        //check server restart, if restart , dead server path in zk should be delete
        if (opType.equals(DELETE_OP)) {
            removeDeadServerByHost(host, type);

        } else if (opType.equals(ADD_OP)) {
            String deadServerPath = getDeadZNodeParentPath() + SINGLE_SLASH + type + UNDERLINE + host;
            if (!isExisted(deadServerPath)) {
                //add dead server info to zk dead server path : /dead-servers/

                persist(deadServerPath, (type + UNDERLINE + host));

                logger.info("{} server dead , and {} added to zk dead server path success",
                        nodeType, node);
            }
        }

    }

    /**
     * get host ip:port, string format: parentPath/ip:port
     *
     * @param path path
     * @return host ip:port, string format: parentPath/ip:port
     */
    public String getHostByEventDataPath(String path) {
        if (org.apache.commons.lang.StringUtils.isEmpty(path)) {
            logger.error("empty path!");
            return "";
        }
        String[] pathArray = path.split(SINGLE_SLASH);
        if (pathArray.length < 1) {
            logger.error("parse ip error: {}", path);
            return "";
        }
        return pathArray[pathArray.length - 1];

    }
    /**
     * remove dead server by host
     *
     * @param host host
     * @param serverType serverType
     */
    public void removeDeadServerByHost(String host, String serverType) {
        List<String> deadServers = getChildrenKeys(getDeadZNodeParentPath());
        for (String serverPath : deadServers) {
            if (serverPath.startsWith(serverType + UNDERLINE + host)) {
                String server = getDeadZNodeParentPath() + SINGLE_SLASH + serverPath;
                remove(server);
                logger.info("{} server {} deleted from zk dead server path success", serverType, host);
            }
        }
    }

    /**
     * check the zookeeper node already exists
     *
     * @param host host
     * @param nodeType zookeeper node type
     * @return true if exists
     */
    public boolean checkNodeExists(String host, NodeType nodeType) {
        String path = getNodeParentPath(nodeType);
        if (org.apache.commons.lang.StringUtils.isEmpty(path)) {
            logger.error("check zk node exists error, host:{}, zk node type:{}",
                    host, nodeType);
            return false;
        }
        Map<String, String> serverMaps = getServerMaps(nodeType);
        for (String hostKey : serverMaps.keySet()) {
            if (hostKey.contains(host)) {
                return true;
            }
        }
        return false;
    }
    /**
     * @param nodeType zookeeper node type
     * @return get zookeeper node parent path
     */
    public String getNodeParentPath(NodeType nodeType) {
        String path = "";
        switch (nodeType) {
            case MASTER:
                return getMasterNodeParentPath();
            case WORKER:
                return getWorkerNodeParentPath();
            case DEAD_SERVER:
                return getDeadNodeParentPath();
            default:
                break;
        }
        return path;
    }
    /**
     * @return get worker node parent path
     */
    protected String getWorkerNodeParentPath() {
        return Constants.REGISTRY_SCHEDULER_WORKERS;
    }

    /**
     * @return get master node parent path
     */
    protected String getMasterNodeParentPath() {
        return Constants.REGISTRY_SCHEDULER_MASTERS;
    }

    /**
     * @return get dead server node parent path
     */
    protected String getDeadNodeParentPath() {
        return Constants.REGISTRY_SCHEDULER_DEAD_SERVERS;
    }
    /**
     * get server list map.
     *
     * @param nodeType zookeeper node type
     * @return result : {host : resource info}
     */
    public Map<String, String> getServerMaps(NodeType nodeType) {
        Map<String, String> serverMap = new HashMap<>();
        try {
            String path = getNodeParentPath(nodeType);
            List<String> serverList = getServerNodes(nodeType);
            for (String server : serverList) {
                String host = server;
                serverMap.putIfAbsent(host, get(path + SINGLE_SLASH + server));
            }
        } catch (Exception e) {
            logger.error("get server list failed", e);
        }

        return serverMap;
    }
    /**
     * get server nodes.
     *
     * @param nodeType registry node type
     * @return result : list<node>
     */
    public List<String> getServerNodes(NodeType nodeType) {
        String path = getNodeParentPath(nodeType);
        List<String> serverList = getChildrenKeys(path);
        return serverList;
    }
    /**
     * get active master num
     *
     * @return active master number
     */
    public int getActiveMasterNum() {
        List<String> childrenList = new ArrayList<>();
        try {
            // read master node parent path from conf
            if (isExisted(getNodeParentPath(NodeType.MASTER))) {
                childrenList = getChildrenKeys(getNodeParentPath(NodeType.MASTER));
            }
        } catch (Exception e) {
            logger.error("getActiveMasterNum error", e);
        }
        return childrenList.size();
    }
}

